package com.jackrain.nea.clould.mq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.util.TypeUtils;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.impl.util.MsgConvertUtil;
import com.jackrain.nea.clould.services.CallInterfaceService;
import com.jackrain.nea.exception.NDSException;
import com.jackrain.nea.util.ApplicationContextHandle;
import com.jackrain.nea.util.Tools;
import com.jackrain.nea.util.ValueHolder;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.Serializable;
import java.io.StringWriter;
import java.util.HashMap;

import static com.aliyun.openservices.ons.api.Action.CommitMessage;
import static com.aliyun.openservices.ons.api.Action.ReconsumeLater;

/**
 * Created by nathan on 2017/5/23.
 */
@Slf4j
public class MQMessageListener implements MessageListener {
    @Override
    public Action consume(Message message, ConsumeContext context) {
        try {
            ValueHolder valueHolder = mqConsume(message, context);
            if(valueHolder.toJSONObject().getInteger("code") == -2){
                log.info("consume later ->" + valueHolder.toJSONObject());
                return ReconsumeLater;
            }
            return CommitMessage;
        } catch (Exception e) {
            StringWriter sw = new StringWriter();
            PrintWriter pw = new PrintWriter(sw);
            e.printStackTrace(pw);
            log.error(sw.toString());
            return ReconsumeLater;
        }

    }

    public ValueHolder mqConsume(Message message, ConsumeContext context) throws Exception {

        log.debug("mqConsume-> callinterface");

        CallInterfaceService callInterfaceService = ApplicationContextHandle.getBean(CallInterfaceService.class);

        Serializable ser = null;
        ser = MsgConvertUtil.objectDeserialize(message.getBody());

        HashMap hashMap = new HashMap();
        hashMap.put("MSGKEY", message.getKey());


        if (ser instanceof JSONArray) {
            log.debug("mqConsume ser->JSONArray");
            JSONArray paramArray = JSONArray.parseArray(ser.toString());
            hashMap.put("PARAM", paramArray);
        }else if(ser instanceof  JSONObject){
            log.debug("mqConsume ser->JSONObject");
            JSONObject paramObject = JSONObject.parseObject(ser.toString());
            hashMap.put("PARAM", paramObject);
        }else{
            log.debug("mqConsume ser->typeof is not json");
            JSONObject paramObject = JSONObject.parseObject((String)ser);
            hashMap.put("PARAM", paramObject);
        }

        ValueHolder valueHolder = callInterfaceService.execute(hashMap);
        return valueHolder;


    }
}
